package com.angelasp.main;

/**
 * @Author yangb
 * @Date 2022/8/13 17:13
 * @Version 1.0
 * //
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.angelasp.service.EsLogAppendServer;
import com.angelasp.util.DateUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class CustomSink extends RichSinkFunction {

    /**
     * 批量处理数据参数
     */
    private final List<Map<String, Object>> list = new ArrayList<>();

    @Override
    public void invoke(Object value, Context context) throws Exception {

        try {
            JSONObject jsonObject = JSON.parseObject(value.toString());
            JSONObject jsonAfter = JSON.parseObject(jsonObject.get("after").toString());
            if (jsonObject != null) {
                Map<String, Object> map = new HashMap<String, Object>();
                for (Map.Entry<String, Object> entry : jsonAfter.entrySet()) {
                    //这里处理一下日期变成时间戳问题,以下进行遍历执行
                    map.put(entry.getKey(), entry.getValue());
                }
                list.add(map);
            }
            if (list.size() > 0) {
                saveElasticSearch(list);
                list.clear();
            }

        } catch (Exception ex) {
            log.info(DateUtils.getDate() + "---" + ex.toString());
        }


    }


    /**
     * 保存到华为ES 中
     * @param list
     */
    public void saveElasticSearch(List<Map<String, Object>> list) {
        try {
            EsLogAppendServer esLogAppendServer = new EsLogAppendServer();
            //我们每天生成一个索引
            String indexName = "my_index_" + DateUtils.getDate();
            //检查创建模板，如果没有就创建，有直接跳过
            esLogAppendServer.CreateMapping(indexName);
            if (list.size() > 0) {
                //添加到ES
                esLogAppendServer.addElasticsearchData(indexName, list);
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

}



